package com.test.test1;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.LinkedList;

public class Reducer1 extends Reducer<LongWritable, Text, Text, Text> {

    final Text k = new Text();
    final Text v = new Text();
    // 创建队列
    LinkedList<BigDecimal> list = new LinkedList<>();

    @Override
    protected void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            String line = value.toString();
            String[] split = line.split(",");
            String date = split[0];
            // 入队
            list.addLast(new BigDecimal(split[1]));
            // 当队列满20，计算20日MA
            if (list.size() == 20) {
                // 计算平均值并保留8位小数
                BigDecimal divide = list.stream().reduce(BigDecimal.ZERO, BigDecimal::add)
                        .divide(new BigDecimal(20), 8, RoundingMode.HALF_UP);
                // 出队
                list.removeFirst();
                k.set(date + ",");
                v.set(divide.toString());
                context.write(k, v);
            }
        }
    }
}
